package com.lagou.dw.flume.interceptors;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Charsets;
import com.google.common.base.Strings;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.junit.Test;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class CustomizedTimeInterceptor implements Interceptor{
    private static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        String body = new String(event.getBody(), Charsets.UTF_8);
        Map<String, String> headerMap = event.getHeaders();
        String[] bodyArr = body.split("\\s+");

        try {
            String jsonStr = bodyArr[6];
            if(Strings.isNullOrEmpty(jsonStr)){
                return null;
            }
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            String timeStampString = jsonObject.getJSONObject("app_active").getString("time");
            long timeStamp = Long.parseLong(timeStampString);
            String date = formatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(timeStamp), ZoneId.systemDefault()));
            headerMap.put("logtime", date);
        } catch (Exception e) {
            headerMap.put("logtime", "unknown");
//            e.printStackTrace();
        }
        event.setHeaders(headerMap);

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> result = new ArrayList<>();
        for(Event e: events){
            Event out = intercept(e);
            if(null != out){
                result.add(out);
            }
        }
        return result;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new CustomizedTimeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

    @Test
    public void test1(){
        String str = "2020-08-20 11:56:08.703 [main] INFO  com.lagou.ecommerce.AppStart - {\"app_active\":{\"name\":\"app_active\",\"json\":{\"entry\":\"1\",\"action\":\"0\",\"error_code\":\"0\"},\"time\":1595321409443},\"attr\":{\"area\":\"温州\",\"uid\":\"2F10092A9999\",\"app_v\":\"1.1.13\",\"event_type\":\"common\",\"device_id\":\"1FB872-9A1009999\",\"os_type\":\"7.43\",\"channel\":\"CE\",\"language\":\"chinese\",\"brand\":\"iphone-5\"}}";
        SimpleEvent simpleEvent = new SimpleEvent();
        simpleEvent.setBody(str.getBytes());
        intercept(simpleEvent);
        System.out.println(simpleEvent.getHeaders());
    }
}
